/**
 * kafka的工具类
 * @author 冯永泽
 * @date 2019/12/3
 */

const kafkaNode = require('kafka-node');
class kafka {
    constructor (ip, topic, options) {
        this.ip = ip;
        this.topic = topic;
        this.options = options || {
            autoCommit: true,
            fetchMaxWaitMs: 1000,
            fetchMaxBytes: 1024 * 1024
        };
        this.client = new kafkaNode.KafkaClient({
            kafkaHost: this.ip
        })
        const topics = this.topic;
        this.consumer = new kafkaNode.Consumer(this.client, topics, this.options);
    }
    /**
     * kafka消息接收时回调函数
     * @param {Function} messageFunction 
     */
    message (messageFunction) {
        this.consumer.on('message', messageFunction);
    }
    /**
     * client出错时回调函数
     * @param {Function} errorFunction 
     */
    error (errorFunction) {
        this.consumer.on('error', errorFunction);
    }
}
module.exports = kafka;